package test

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

/**
  * Created by wanghongli on 2021/2/6.
  *
  *
  */
object sparkonholo {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName("spark_sql_ddl")
      .master("local[*]")
      //      .config("spark.sql.catalogImplementation", "odps")
      //      .config("spark.hadoop.odps.end.point","http://service.cn.maxcompute.aliyun.com/api")
      //      .config("spark.hadoop.odps.runtime.end.point","http://service.cn.maxcompute.aliyun-inc.com/api")
      //      .config("spark.hadoop.odps.access.id","LTAI4Fzm6tCPCiAutd9tXicj")
      //      .config("spark.hadoop.odps.access.key","IPiUVIlP1arYKBX3hETT0MdSkE4qeQ")
      //      .config("spark.hadoop.odps.end.point", "http://service.cn.maxcompute.aliyun.com/api")
      .config("spark.sql.broadcastTimeout", "1200")
      .config("spark.sql.crossJoin.enabled", "true")
      .config("odps.exec.dynamic.partition.mode", "nonstrict")
      .config("spark.hadoop.odps.project.name", "O2O_DataWorks_dev")
      .getOrCreate()
    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    //    val df: DataFrame = spark.sql(
    //      """
    //      select * from dw_jd_daojia_good_detail_2020 where dt='202010'
    //      """.stripMargin)

    val df = spark.read.orc("s3a://o2o-dataproces-group/wang_hongli/taobao/nianhuojie/2020/1_clear/part-00000*")
    df.registerTempTable("df")
    //   val tmall_r = df.sqlContext.sql(
    //      """
    //        select
    //        |cast(shopurl as string) as shopurl,
    //        |cast(sellertype as string) as sellertype,
    //        |cast(creditlevel as string) as creditlevel,
    //        |cast(good_id as string) as good_id,
    //        |cast(is_oversea as string) as is_oversea,
    //        |cast(is_newproduct as string) as is_newproduct,
    //        |cast(userid as string) as userid,
    //        |cast(rootcategoryid as string) as rootcategoryid,
    //        |cast(goodratepercentage as string) as goodratepercentage,
    //        |cast(images as string) as images,
    //        |cast(shoptype as string) as shoptype,
    //        |cast(sellernick as string) as sellernick,
    //        |cast(title as string) as title,
    //        |cast(bc_type as string) as bc_type,
    //        |cast(fans as string) as fans,
    //        |cast(allitemcount as string) as allitemcount,
    //        |cast(originitem as string) as originitem,
    //        |cast(favcount as string) as favcount,
    //        |cast(categoryid as string) as categoryid,
    //        |cast(original_cost as string) as original_cost,
    //        |cast(base_info as string) as base_info,
    //        |cast(is_showlive as string) as is_showlive,
    //        |cast(delivery_from as string) as delivery_from,
    //        |cast(evaluates as string) as evaluates,
    //        |cast(shopname as string) as shopname,
    //        |cast(starts as string) as starts,
    //        |cast(originproduction as string) as originproduction,
    //        |cast(shopid as string) as shopid,
    //        |cast(origincountryname as string) as origincountryname,
    //        |cast(brandname as string) as brandname,
    //        |cast(detailcategoryname as string) as detailcategoryname,
    //        |cast(brandvalueid as string) as brandvalueid,
    //        |cast(goodurl as string) as goodurl,
    //        |cast(lastcategoryid as string) as lastcategoryid,
    //        |cast(quantity as string) as quantity,
    //        |cast(sellcount as string) as sellcount,
    //        |cast(pricetext as string) as pricetext,
    //        |cast(commentcount as string) as commentcount,
    //        |cast(salesamount as string) as salesamount,
    //        |cast(sh_categoryname as string) as sh_categoryname,
    //        |cast(fy_categoryname as string) as fy_categoryname,
    //        |cast(dt as string) as dt,
    //        |cast(platform_id as string) as platform_id
    //        | from
    //        |df
    //        |""".stripMargin)

    //    val jd_r = df.sqlContext.sql(
    //      """
    //        |select
    //        |cast(shopUrl as string) as shopUrl,
    //        |cast(crawl_date as string) as crawl_date,
    //        |cast(good_id as string) as good_id,
    //        |cast(is_oversea as string) as is_oversea,
    //        |cast(is_newProduct as string) as is_newProduct,
    //        |cast(rootCategoryId as string) as rootCategoryId,
    //        |cast(goodRatePercentage as string) as goodRatePercentage,
    //        |cast(subCategoryName as string) as subCategoryName,
    //        |cast(images as string) as images,
    //        |cast(shopType as string) as shopType,
    //        |cast(title as string) as title,
    //        |cast(fans as string) as fans,
    //        |cast(allItemCount as string) as allItemCount,
    //        |cast(rootCategoryName as string) as rootCategoryName,
    //        |cast(originItem as string) as originItem,
    //        |cast(categoryId as string) as categoryId,
    //        |cast(categoryName as string) as categoryName,
    //        |cast(original_cost as string) as original_cost,
    //        |cast(Base_Info as string) as Base_Info,
    //        |cast(delivery_from as string) as delivery_from,
    //        |cast(evaluates as string) as evaluates,
    //        |cast(subCategoryId as string) as subCategoryId,
    //        |cast(shopName as string) as shopName,
    //        |cast(starts as string) as starts,
    //        |cast(originProduction as string) as originProduction,
    //        |cast(shopId as string) as shopId,
    //        |cast(originCountryName as string) as originCountryName,
    //        |cast(brandName as string) as brandName,
    //        |cast(detailCategoryName as string) as detailCategoryName,
    //        |cast(brandValueId as string) as brandValueId,
    //        |cast(goodUrl as string) as goodUrl,
    //        |cast(sellCount as string) as sellCount,
    //        |cast(priceText as string) as priceText,
    //        |cast(commentCount as string) as commentCount,
    //        |cast(is_selfSupport as string) as is_selfSupport,
    //        |cast(salesAmount as string) as salesAmount,
    //        |cast(sh_categoryname as string) as sh_categoryname,
    //        |cast(fy_categoryname as string) as fy_categoryname,
    //        |cast(dt as string) as dt ,
    //        |cast(platform_id as string) as platform_id,
    //        |cast(lastCategoryId as string) as lastCategoryId
    //        |from df
    //        |""".stripMargin)

    val taobao_r = df.sqlContext.sql(
      """
        |select cast(shopurl as string) shopurl,
        |cast(sellertype as string) sellertype,
        |cast(creditlevel as string) creditlevel,
        |cast(good_id as string) good_id,
        |cast(is_oversea as string) is_oversea,
        |cast(userid as string) userid,
        |cast(rootcategoryid as string) rootcategoryid,
        |cast(goodratepercentage as string) goodratepercentage,
        |cast(images as string) images,
        |cast(shoptype as string) shoptype,
        |cast(sellernick as string) sellernick,
        |cast(title as string) title,
        |cast(bc_type as string) bc_type,
        |cast(fans as string) fans,
        |cast(allitemcount as string) allitemcount,
        |cast(originitem as string) originitem,
        |cast(favcount as string) favcount,
        |cast(categoryid as string) categoryid,
        |cast(original_cost as string) original_cost,
        |cast(base_info as string) base_info,
        |-- cast(is_showlive as string) is_showlive,
        |cast(delivery_from as string) delivery_from,
        |cast(evaluates as string) evaluates,
        |cast(shopname as string) shopname,
        |cast(starts as string) starts,
        |cast(originproduction as string) originproduction,
        |cast(shopid as string) shopid,
        |cast(brandname as string) brandname,
        |cast(brandvalueid as string) brandvalueid,
        |cast(goodurl as string) goodurl,
        |cast(lastcategoryid as string) lastcategoryid,
        |cast(quantity as string) quantity,
        |cast(sellcount as string) sellcount,
        |cast(pricetext as string) pricetext,
        |cast(commentcount as string) commentcount,
        |cast(salesamount as string) salesamount,
        |cast(sh_categoryname as string) sh_categoryname,
        |cast(fy_categoryname as string) fy_categoryname,
        |cast(dt as string) dt,
        |cast(platform_id as string) platform_id
        |from df
        |""".stripMargin)


    val out: RDD[Row] = taobao_r.rdd.map(row => Row(
      row.get(0)
      ,row.get(1)
      ,row.get(2)
      ,row.get(3)
      ,row.get(4)
      ,row.get(5)
      ,row.get(6)
      ,row.get(7)
      ,row.get(8)
      ,row.get(9)
      ,row.get(10)
      ,row.get(11)
      ,row.get(12)
      ,row.get(13)
      ,row.get(14)
      ,row.get(15)
      ,row.get(16)
      ,row.get(17)
      ,row.get(18)
      ,row.get(19)
      ,row.get(20)
      ,row.get(21)
      ,row.get(22)
      ,row.get(23)
      ,row.get(24)
      ,row.get(25)
      ,row.get(26)
      ,row.get(27)
      ,row.get(28)
      ,row.get(29)
      ,row.get(30)
      ,row.get(31)
      ,row.get(32)
      ,row.get(33)
      ,row.get(34)
      ,row.get(35)
      ,row.get(36)
      ,row.get(37)
      ,row.get(38)
      //,row.get(39)


    ))
    //    val schema_jd = StructType(
    //      Seq(
    //        StructField("shopUrl",StringType,true),
    //        StructField("crawl_date",StringType,true),
    //        StructField("good_id",StringType,true),
    //        StructField("is_oversea",StringType,true),
    //        StructField("is_newProduct",StringType,true),
    //        StructField("rootCategoryId",StringType,true),
    //        StructField("goodRatePercentage",StringType,true),
    //        StructField("subCategoryName",StringType,true),
    //        StructField("images",StringType,true),
    //        StructField("shopType",StringType,true),
    //        StructField("title",StringType,true),
    //        StructField("fans",StringType,true),
    //        StructField("allItemCount",StringType,true),
    //        StructField("rootCategoryName",StringType,true),
    //        StructField("originItem",StringType,true),
    //        StructField("categoryId",StringType,true),
    //        StructField("categoryName",StringType,true),
    //        StructField("original_cost",StringType,true),
    //        StructField("Base_Info",StringType,true),
    //        StructField("delivery_from",StringType,true),
    //        StructField("evaluates",StringType,true),
    //        StructField("subCategoryId",StringType,true),
    //        StructField("shopName",StringType,true),
    //        StructField("starts",StringType,true),
    //        StructField("originProduction",StringType,true),
    //        StructField("shopId",StringType,true),
    //        StructField("originCountryName",StringType,true),
    //        StructField("brandName",StringType,true),
    //        StructField("detailCategoryName",StringType,true),
    //        StructField("brandValueId",StringType,true),
    //        StructField("goodUrl",StringType,true),
    //        StructField("sellCount",StringType,true),
    //        StructField("priceText",StringType,true),
    //        StructField("commentCount",StringType,true),
    //        StructField("is_selfSupport",StringType,true),
    //        StructField("salesAmount",StringType,true),
    //        StructField("sh_categoryname",StringType,true),
    //        StructField("fy_categoryname",StringType,true),
    //        StructField("dt",StringType,true),
    //        StructField("platform_id",StringType,true),
    //        StructField("lastCategoryId",StringType,true)
    //      )
    //    )

    val schema_tao = StructType(
      Seq(
        StructField("shopUrl",StringType,true),
        StructField("sellertype",StringType,true),
        StructField("creditlevel",StringType,true),
        StructField("good_id",StringType,true),
        StructField("is_oversea",StringType,true),
        //StructField("is_newProduct",StringType,true),
        StructField("userid",StringType,true),
        StructField("rootCategoryId",StringType,true),
        StructField("goodRatePercentage",StringType,true),
        StructField("images",StringType,true),
        StructField("shopType",StringType,true),
        StructField("sellernick",StringType,true),
        StructField("title",StringType,true),
        StructField("bc_type",StringType,true),
        StructField("fans",StringType,true),
        StructField("allItemCount",StringType,true),
        StructField("originItem",StringType,true),
        StructField("favcount",StringType,true),
        StructField("categoryId",StringType,true),
        StructField("original_cost",StringType,true),
        StructField("base_info",StringType,true),
        // StructField("is_showlive",StringType,true),
        StructField("delivery_from",StringType,true),
        StructField("evaluates",StringType,true),
        StructField("shopName",StringType,true),
        StructField("starts",StringType,true),
        StructField("originProduction",StringType,true),
        StructField("shopId",StringType,true),
        //  StructField("originCountryName",StringType,true),
        StructField("brandName",StringType,true),
        //  StructField("detailCategoryName",StringType,true),
        StructField("brandValueId",StringType,true),
        StructField("goodUrl",StringType,true),
        StructField("lastCategoryId",StringType,true),
        StructField("quantity",StringType,true),
        StructField("sellCount",StringType,true),
        StructField("priceText",StringType,true),
        StructField("commentCount",StringType,true),
        StructField("salesAmount",StringType,true),
        StructField("sh_categoryname",StringType,true),
        StructField("fy_categoryname",StringType,true),
        StructField("dt",StringType,true),
        StructField("platform_id",StringType,true)
      )
    )


    val res: DataFrame = spark.createDataFrame(out,schema_tao)


    //    val schema_tmall = StructType(
    //            Seq(
    //              StructField("shopUrl",StringType,true),
    //              StructField("sellertype",StringType,true),
    //              StructField("creditlevel",StringType,true),
    //              StructField("good_id",StringType,true),
    //              StructField("is_oversea",StringType,true),
    //              StructField("is_newProduct",StringType,true),
    //              StructField("userid",StringType,true),
    //              StructField("rootCategoryId",StringType,true),
    //              StructField("goodRatePercentage",StringType,true),
    //              StructField("images",StringType,true),
    //              StructField("shopType",StringType,true),
    //              StructField("sellernick",StringType,true),
    //              StructField("title",StringType,true),
    //              StructField("bc_type",StringType,true),
    //              StructField("fans",StringType,true),
    //              StructField("allItemCount",StringType,true),
    //              StructField("originItem",StringType,true),
    //              StructField("favcount",StringType,true),
    //              StructField("categoryId",StringType,true),
    //              StructField("original_cost",StringType,true),
    //              StructField("base_info",StringType,true),
    //              StructField("is_showlive",StringType,true),
    //              StructField("delivery_from",StringType,true),
    //              StructField("evaluates",StringType,true),
    //              StructField("shopName",StringType,true),
    //              StructField("starts",StringType,true),
    //              StructField("originProduction",StringType,true),
    //              StructField("shopId",StringType,true),
    //              StructField("originCountryName",StringType,true),
    //              StructField("brandName",StringType,true),
    //              StructField("detailCategoryName",StringType,true),
    //              StructField("brandValueId",StringType,true),
    //              StructField("goodUrl",StringType,true),
    //              StructField("lastCategoryId",StringType,true),
    //                StructField("quantity",StringType,true),
    //              StructField("sellCount",StringType,true),
    //              StructField("priceText",StringType,true),
    //              StructField("commentCount",StringType,true),
    //              StructField("salesAmount",StringType,true),
    //              StructField("sh_categoryname",StringType,true),
    //              StructField("fy_categoryname",StringType,true),
    //              StructField("dt",StringType,true),
    //              StructField("platform_id",StringType,true)
    //
    //            )
    //          )
    //   val res: DataFrame = spark.createDataFrame(out,schema_tmall)


    res.write
      .format("datahub")//必须配置为DataHub。
      .option("access.key.id", "LTAI4Fzm6tCPCiAutd9tXicj")//当前阿里云账号的AccessKey ID。
      .option("access.key.secret", "IPiUVIlP1arYKBX3hETT0MdSkE4qeQ")//当前阿里云账号的AccessKey Secret。
      .option("endpoint", "hgprecn-cn-m7r1xxicx002-hh-cn-beijing.hologres.aliyuncs.com:80")//当前Hologres实时数据API的Endpoint，即从步骤3的执行结果中获取的Endpoint。
      .option("project", "pdb")//当前Hologres的数据库名称。
      //      .option("topic", "jd_daojia_test_tab") //  当前Hologres用于接收数据的表名称。
      //      .option("schema", "wusonghua") //  当前Hologres用于接收数据的表名称。
      //      .option("topic", "jd_daojia_test_tab") //  当前Hologres用于接收数据的表名称。
      .option("topic", "dwd_year_taobao_good_base_mf") //  当前Hologres用于接收数据的表名称。
      .option("batch.size", "100") //配置批量导入数据的条数，此处示例配置为100条。
      .option("decimal.precision", 38)
      .option("decimal.scale", 18)
      .save()
    spark.stop()






    //      res.write
    //      .format("datahub")//必须配置为DataHub。
    //      .option(DatahubSourceProvider.OPTION_KEY_ACCESS_ID, "LTAI4Fzf9VpMw6WvPzfdzATD")//阿里云账号的AccessKey ID。
    //      .option(DatahubSourceProvider.OPTION_KEY_ACCESS_KEY, "4OlpmCEWsVAg7yR2uAdfW57LYNx453")//阿里云账号的Accesskey Secret。
    //      .option(DatahubSourceProvider.OPTION_KEY_ENDPOINT, "hgprecn-cn-m7r1xxicx002-hh-cn-beijing.hologres.aliyuncs.com:80")//Hologres实时数据API的Endpoint和Port。
    //      .option(DatahubSourceProvider.OPTION_KEY_PROJECT, "testdb")//Hologres的数据库名称。
    //      .option(DatahubSourceProvider.OPTION_KEY_TOPIC, "jd_daojia_test_part_tab")//Hologres用于接收数据的表名称，示例接收数据的表为tb001。
    //      .option("decimal.precision", 38)
    //      .option("decimal.scale", 18)
    //      .option("batch.size", "100")//配置批量导入数据的条数，此处示例配置为100条。
    //      .save()
    //      intranet,hgprecn-cn-m7r1xxicx002-hh-cn-beijing-internal.hologres.aliyuncs.com:80;
    //      vpc,hgprecn-cn-m7r1xxicx002-hh-cn-beijing-vpc.hologres.aliyuncs.com:80;
    //      internet,hgprecn-cn-m7r1xxicx002-hh-cn-beijing.hologres.aliyuncs.com:80







  }





}
